package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.KafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-15 18:59
 */
public class DwdTrafficLogApp {


    public static final String ONE = "1";

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(4);
        //设置检查点

        //定义topic
        String topicName = "topic_log";

        String groupId = "dwd_traffic_base_log";

        //获取kafka中的数据
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getFlinkKafkaConsumer(topicName, groupId);

        // 添加数据源
        DataStreamSource<String> kakfaDataStream = env.addSource(kafkaConsumer);

        // 定义侧输出流 以及数据的ETL
        OutputTag<String> outputTag = new OutputTag<String>("dirty-data") {
        };
        SingleOutputStreamOperator<JSONObject> clearData = kakfaDataStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(jsonStr);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(outputTag, jsonStr);
                }
            }
        });


        DataStream<String> dirtyData = clearData.getSideOutput(outputTag);

        // 将脏数据发送到kafka中
        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getFlinkKafkaProducer("dirty_data");
        dirtyData.addSink(kafkaProducer);

        //新老访客标志
        KeyedStream<JSONObject, String> keyedStream = clearData
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));


        /**
         * 如果is_new的值为1
         * 如果键控状态为 null，认为本次是该访客首次访问 APP，将日志中 ts 对应的日期更新到状态中，不对 is_new 字段做修改；
         * 如果键控状态不为 null，且首次访问日期不是当日，说明访问的是老访客，将 is_new 字段置为 0；
         * 如果键控状态不为 null，且首次访问日期是当日，说明访问的是新访客，不做操作；
         * 如果 is_new 的值为 0
         * 如果键控状态为 null，说明访问 APP 的是老访客但本次是该访客的页面日志首次进入程序。当前端新老访客状态标记丢失时，日志进入程序被判定为新访客，Flink 程序就可以纠正被误判的访客状态标记，只要将状态中的日期设置为今天之前即可。本程序选择将状态更新为昨日；
         * 如果键控状态不为 null，说明程序已经维护了首次访问日期，不做操作。
         */
        SingleOutputStreamOperator<JSONObject> fixDataStream =
                keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            //定义状态变量
            ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState =
                        getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {

                String isNew = jsonObj.getJSONObject("common").getString("is_new");
                Long ts = jsonObj.getLong("ts");
                String date = DateFormatUtil.toDate(ts);
                if (ONE.equals(isNew)) {
                    if (valueState.value() == null) {
                        valueState.update(date);
                    } else {
                        String firstDate = valueState.value();
                        // 获取当前时间
                        if (!firstDate.equals(date)) {
                            // 有访问，访问日期不是状态变量存的，
                            jsonObj.getJSONObject("common").put("is_new", 0);
                        }
                    }
                } else {
                    String firstDate = valueState.value();
                    if (firstDate == null) {
                        valueState.update(DateFormatUtil.toDate(ts - 24 * 3600 * 1000));
                    }
                }
                out.collect(jsonObj);
            }
        });
        // 数据分流
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        //将不同数据写入不同的日志中
        SingleOutputStreamOperator<String> triageData = fixDataStream
                .process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                // 6.2.1 收集错误数据
                JSONObject error = jsonObj.getJSONObject("err");
                if (error != null) {
                    ctx.output(errorTag, jsonObj.toJSONString());
                    // 剔除 "err" 字段
                    jsonObj.remove("err");
                }

                // 6.2.2 收集启动数据
                JSONObject start = jsonObj.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, jsonObj.toJSONString());
                } else {
                    // 获取 "page" 字段
                    JSONObject page = jsonObj.getJSONObject("page");
                    // 获取 "common" 字段
                    JSONObject common = jsonObj.getJSONObject("common");
                    // 获取 "ts"
                    Long ts = jsonObj.getLong("ts");

                    // 6.2.3 收集曝光数据
                    JSONArray displays = jsonObj.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            ctx.output(displayTag, displayObj.toJSONString());
                        }
                    }

                    // 6.2.4 收集动作数据
                    JSONArray actions = jsonObj.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("common", common);
                            actionObj.put("page", page);
                            ctx.output(actionTag, actionObj.toJSONString());
                        }
                    }

                    // 6.2.5 收集页面数据
                    jsonObj.remove("displays");
                    jsonObj.remove("actions");
                    out.collect(jsonObj.toJSONString());
                }
            }
        });

        //输出日志
        triageData.getSideOutput(actionTag).print("行动日志>>>>");
        triageData.getSideOutput(displayTag).print("曝光日志>>>>");
        triageData.getSideOutput(errorTag).print("错误日志>>>>");
        triageData.getSideOutput(startTag).print("启动日志>>>>>");
        triageData.print("主流数据页面日志");

        //发送kafka对应的主题中
        // 7.1 提取各侧输出流
        DataStream<String> startDataStream = triageData.getSideOutput(startTag);
        DataStream<String> displayDataStream = triageData.getSideOutput(displayTag);
        DataStream<String> actionDataStream = triageData.getSideOutput(actionTag);
        DataStream<String> errorDataStream = triageData.getSideOutput(errorTag);


        // 7.2 定义不同日志输出到 Kafka 的主题名称
        String pageTopic = "dwd_traffic_page_log";
        String startTopic = "dwd_traffic_start_log";
        String displayTopic = "dwd_traffic_display_log";
        String actionTopic = "dwd_traffic_action_log";
        String errorTopic = "dwd_traffic_error_log";

        triageData.addSink(KafkaUtil.getFlinkKafkaProducer(pageTopic));
        startDataStream.addSink(KafkaUtil.getFlinkKafkaProducer(startTopic));
        displayDataStream.addSink(KafkaUtil.getFlinkKafkaProducer(displayTopic));
        actionDataStream.addSink(KafkaUtil.getFlinkKafkaProducer(actionTopic));
        errorDataStream.addSink(KafkaUtil.getFlinkKafkaProducer(errorTopic));

        //执行任务
        env.execute();
    }
}
